package com.stormragetech.sylwanas.process;

import com.stormragetech.sylwanas.common.MessageWrapper;
import com.stormragetech.sylwanas.common.QueueInfo;
import com.stormragetech.sylwanas.common.ReturnInfo;
import com.stormragetech.sylwanas.constant.Constants;
import com.stormragetech.sylwanas.converter.FastjsonMessageConverter;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * description: //TODO
 *
 * @author: Xue Bin
 * @version: 1.0.0-SNAPSHOT
 */
@Log4j2
public class MessageSenderBuilder implements MessageSender {

    private QueueInfo info;
    private ConnectionFactory factory;
    private RetryCache cache;
    private RabbitTemplate template;

    MessageSenderBuilder(QueueInfo info, ConnectionFactory factory, RetryCache cache) {
        this.info = info;
        this.factory = factory;
        this.cache = cache;
        this.template = initTemplate();
    }

    private RabbitTemplate initTemplate() {

        // init rabbit templete
        //
        final RabbitTemplate template = new RabbitTemplate(factory);
        template.setMandatory(true);
        template.setExchange(info.getExchange());
        template.setRoutingKey(info.getRoutingKey());
        template.setMessageConverter(new FastjsonMessageConverter()); // set converter

        // set confirm callback
        // message to exchange
        //
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                cache.getSMap().get(correlationData.getId()).setConfirmCode(-1);
                log.info("[ confirm call back ] send message [ {} ] to exchange failed, cause: {} ", correlationData.getId(), cause);
//                log.info("[ confirm call back ] send message [ {} ] to exchange count [ {} ] failed, cause: {} ", correlationData.getId(), cache.getId(), cause);
            } else {
                cache.getSMap().get(correlationData.getId()).setConfirmCode(1);
                log.info("[ confirm call back ] send message [ {} ] to exchange success !!!", correlationData.getId());
            }
        });

        // set return callback
        // message from exchange to queue
        //
        template.setReturnCallback((message, replyCode, replyText, rExchange, rRoutingKey) -> {
            cache.getSMap().get(message.getMessageProperties().getCorrelationId()).setReturnCode(-1);
            log.info("[ return call back ] send message to queue failed, cause: {} ", replyText);
//            log.info("[ return call back ] send message to queue count [ {} ] failed, cause: {} ", cache.getId(), replyText);
        });

        return template;
    }

    @Override
    public ReturnInfo send(Object message) {

        MessageWrapper wrapper;
        if (!(message instanceof MessageWrapper)) {
            wrapper = new MessageWrapper(message, this);
            cache.add(wrapper, Constants.SEND_MSG_WRAPPER);
        } else {
            wrapper = (MessageWrapper) message;
        }

        try {
            cache.setTime(wrapper.getId());
            template.correlationConvertAndSend(wrapper, new CorrelationData(wrapper.getId()));
            System.out.println();
            System.out.println("---------- ---------- ---------- ---------- ---------- ----------");
            log.info("[ SEND ] - [ MESSAGE - ID ] - [ {} ]", wrapper.getId());
            System.out.println("---------- ---------- ---------- ---------- ---------- ----------");
            System.out.println();
            return new ReturnInfo(Constants.SUCC, "");
        } catch (AmqpConnectException e) {
            return new ReturnInfo(Constants.CONN_ERR, e.getMessage());
        } catch (Exception e) {
            System.out.println();
            System.out.println("---------- ---------- ---------- ---------- ---------- ----------");
            log.info("[ FAILURE] - [ SEND ] - [ MESSAGE - ID ] - [ {} ] - [ CAUSE ] - [ {} ]", wrapper.getId(), e.getMessage());
            System.out.println("---------- ---------- ---------- ---------- ---------- ----------");
            System.out.println();
            log.error("Producer send message failed, cause: {}", e.getMessage());
            return new ReturnInfo(Constants.FAIL, e.getMessage());
        } finally {
            wrapper.unlock();
        }
    }
}
